package syan.redisp.controller;

import syan.redisp.config.Result;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;

/**
 * @author CHENTAIWEN
 * @create 2022/8/3 15:46
 */
@Slf4j
@RestController
@RequestMapping("/Redis")
public class RedisAdd {

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    /**
     * 存放订单文件
     */
    private static String key="Message:Order:";

    private static long count=1;

    private static long trim=1;
    /**
     * XRANGE key start end [COUNT count]
     * @param channel 生产信息储存流
     * @return 生产者创建的所有消息
     */
    @PostMapping("/add")
    public Result<String> add(String channel) throws Exception{
//        Thread.sleep(1000 * (int)(Math.random() * 10));
        List<MapRecord<String, Object, Object>> redis;
        synchronized (this){
            redis =redisTemplate.opsForStream().read(StreamOffset.fromStart(key+channel));
//            放在添加前面的时候 99条确保最新加入时为100(会删除)
//            Long l=redisTemplate.opsForStream().trim(key + channel, 99);
            if ( redis != null && redis.size()!= 0){
//                trim+=l;
                count=redis.size()+trim;
            }
            Map<String, Object> map = new HashMap<>();
            String id= UUID.randomUUID().toString();
            map.put(String.valueOf(count),id);
            redisTemplate.opsForStream().add(key+channel,map);
            System.out.println(id + "读取到内容" +count +"条");
            return Result.ok("已创建"+count+"条数据;最新数据 "+map);
        }
    }

    /**
     *  XGROUP CREATE mystream consumer-group-name 0-0 0-0 从首部 $尾部开始消费，只接受新消息，当前 Stream 消息会全部忽略。
     *  createGroup默认是尾部需要处理
     * @param channel 生产信息储存流
     * @param consumer 消费组
     * @param status true 0-0 从开始 $ 接收新
     * @return OK
     */
    @PostMapping("/addConsumer")
    public Result<String> addConsumer(String channel, String consumer, Boolean status){
        String redis=null;
        if (status) {
            redis= redisTemplate.opsForStream().createGroup(key + channel, ReadOffset.from("0-0"), consumer);
        }else{
            redis= redisTemplate.opsForStream().createGroup(key + channel, ReadOffset.latest(), consumer);
        }
        return Result.ok(redis);
    }

    /**
     * XREADGROUP GROUP consumer-group-name consumer-name COUNT 1 STREAMS mystream >
     * @param channel  生产信息储存流
     * @param name 消费者
     * @param group 消费组
     * @return 消费者消费的消息
     */
    @PostMapping("/readConsumer")
    public Result<List<MapRecord<String, Object, Object>>> readConsumer(String channel, String name, String group, Integer count){
        Consumer consumer= Consumer.from(group,name);
        List<MapRecord<String, Object, Object>> records;
        if (count!=null && count > 0) {
            records= redisTemplate.opsForStream().read(consumer, StreamReadOptions.empty().count(count).block(Duration.ofSeconds(1)), StreamOffset.create(key + channel, ReadOffset.lastConsumed()));
        }else{
            records= redisTemplate.opsForStream().read(consumer, StreamOffset.create(key + channel, ReadOffset.lastConsumed()));
        }
        return Result.ok(records);
    }
}
